package cn.azzhu.proj

import java.sql.Timestamp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.Set

/**
 * 如果访问量很大怎么办？这里的方法会把所有的PV数据放在窗口里面，然后去重 ===>增量聚合
 * @author azzhu
 * @create 2020-09-24 21:27:00
 */
object UVAgg {
  case class UserBehaviour(userId:Long,
                           itemId:Long,
                           categoryId:Int,
                           behaviour: String,
                           timestamp:Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //为了时间旅行，必须使用事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream = env
      .readTextFile("D:\\bigdata\\flink-learning\\src\\main\\resources\\UserBehavior.csv")
      .map(line => {
        val arr = line.split(",")
        //todo 注意：时间戳单位必须是毫秒
        UserBehaviour(arr(0).toLong,arr(1).toLong,arr(2).toInt,arr(3),arr(4).toLong * 1000)
      })
      .filter(_.behaviour.equals("pv")) //过滤出pv事件
      .assignAscendingTimestamps(_.timestamp) //分配升序时间戳
      .map(r => ("key",r.userId))
      .keyBy(_._1)
      .timeWindow(Time.hours(1))
      .aggregate(new CountAgg,new WindowResult)
      .print()

    env.execute("UVAgg")
  }

  class Agg {
    var count = 0L
    var set = Set[Long]()
  }

  class CountAgg extends AggregateFunction[(String,Long),Agg,Long] {
    override def createAccumulator(): Agg = new Agg

    override def add(value: (String, Long), acc: Agg): Agg = {
      if(!acc.set.contains(value._2)) {
        acc.set += value._2
        acc.count += 1
      }
      acc
    }

    override def getResult(acc: Agg): Long = acc.count

    override def merge(acc: Agg, acc1: Agg): Agg = ???
  }

  //如果滑动窗口是1小时，滑动距离是5秒钟，每小时用户数量是10亿呢？还管用吗？
  //也就是说每小时的uv是10亿，去重完以后Set里面都有10亿个userId
  //每一个userId 是1kb，10亿个userId是多少？1T的数据；每隔5s就会产生1T的数据
  class WindowResult extends ProcessWindowFunction[Long,String,String,TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[Long], out: Collector[String]): Unit = {
      out.collect("窗口结束时间为： "+new Timestamp(context.window.getEnd) + " 的窗口的UV统计值是：" + elements.head)
    }
  }
}
